package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.redis.RedisCacheService;
import com.heima.common.redis.ScheduleConstants;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;

import java.util.*;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
@Transactional
public class TaskServiceImpl implements TaskService {
    //添加任务
    @Override
    public long addTask(Task task) {
        //添加任务到数据库
        boolean addTaskToDb = addTaskToDb(task);
        if (addTaskToDb) {
            addTaskToCache(task);
        }


        //添加任务到redis

        return task.getTaskId();
    }

    //取消任务
    @Override
    public Boolean deleteTask(long taskId) {
        Task task = deleteTaskDb(taskId, ScheduleConstants.EXECUTED);
        if (task != null) {
            deleteTaskCache(task);
            return true;
        }

        return false;
    }

    //消费redis队列
    @Override
    public Task poll(int type, int priority) {
        Task task = null;
        //获取key
        String key = type + "_" + priority;
        //根据key拉取队列
        String pop = redisCacheService.lRightPop(ScheduleConstants.TOPIC + key);
        //去过拉去的队列不为空,转换为task类型
        if (pop != null) {
            task = JSON.parseObject(pop, Task.class);
            //更新数据库

            deleteTaskDb(task.getTaskId(), ScheduleConstants.EXECUTED);
            return task;
        }


        return task;

    }

    //删除Redis队列
    private void deleteTaskCache(Task task) {
        String key = task.getTaskType() + "_" + task.getPriority();
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            redisCacheService.lRemove(ScheduleConstants.TOPIC + key, 0, JSON.toJSONString(task));
        } else {
            redisCacheService.zRemove(ScheduleConstants.FUTURE + key, JSON.toJSONString(task));
        }
    }

    private Task deleteTaskDb(long taskId, int status) {
        //删除数据库任务表
        taskinfoMapper.deleteById(taskId);
        //更改日志表
        TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
        taskinfoLogs.setStatus(status);
        Task task = new Task();
        BeanUtils.copyProperties(taskinfoLogsMapper, task);
        task.setExecuteTime(new Date().getTime());
        return task;
    }

    @Autowired
    TaskinfoMapper taskinfoMapper;
    @Autowired
    TaskinfoLogsMapper taskinfoLogsMapper;

    //添加任务到数据库
    private boolean addTaskToDb(Task task) {
        Boolean falg = false;
        //将任务表保存到数据库
        try {
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(task, taskinfo);
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskinfoMapper.insert(taskinfo);
            //设置taskId
            task.setTaskId(taskinfo.getTaskId());
            //将任务日志表保存到数据库
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(taskinfo, taskinfoLogs);
            taskinfoLogs.setVersion(1);
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            taskinfoLogsMapper.insert(taskinfoLogs);
        } catch (BeansException e) {
            return falg;
        }
        return falg = true;


    }

    @Autowired
    RedisCacheService redisCacheService;

    //添加任务到redis
    private void addTaskToCache(Task task) {

        //设置key名
        String key = task.getTaskType() + "-" + task.getPriority();
        //获取5分钟毫秒值
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        long nextInMillis = calendar.getTimeInMillis();

        //判断执行时间小于当前时间存入list当前队列
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            redisCacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
        }
        //判断执行时间小于当预设时间存入Zset未来数据队列
        if (task.getExecuteTime() <= nextInMillis) {
            redisCacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), nextInMillis);
        }


    }

    //定时刷新
    @Scheduled(cron = "0 */1 * * * ?")
    public void refresh() {
        String token = redisCacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
        if (token != null) {
            log.info("未来数据定时刷新---定时任务");
            //获取所有未来key
            Set<String> futureKeys = redisCacheService.scan(ScheduleConstants.FUTURE + "*");

            for (String futureKey : futureKeys) {
                //获取需要的消费的任务数据
                Set<String> task = redisCacheService.zRange(futureKey, 0, System.currentTimeMillis());
                if (task != null) {
                    String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];
                    redisCacheService.refreshWithPipeline(futureKey, topicKey, task);
                    System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
                }
            }
        }
    }
    //数据库同步到redis
    @Scheduled(cron = "0 */5 * * * ?")
    @PostConstruct
    public void reloadData(){
        // 删除缓存中未来数据集合和当前消费者队列的所有key
        clearCache();
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);
        //查看小于未来10分钟所有对象
        List<Taskinfo> taskinfos = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime, calendar.getTime()));

        //当前执行时间<未来5分值时间
        if (taskinfos != null) {
            for (Taskinfo taskinfo : taskinfos) {
                Task task = new Task();
                BeanUtils.copyProperties(taskinfo, task);
                task.setExecuteTime(taskinfo.getExecuteTime().getTime());
                addTaskToCache(task);
            }
        }

    }
    //删除当前未来任务
    public void clearCache() {
        // 删除缓存中未来数据集合和当前消费者队列的所有key
        Set<String> topicKey = redisCacheService.scan(ScheduleConstants.TOPIC + "*");
        Set<String> futureKey = redisCacheService.scan(ScheduleConstants.FUTURE + "*");
      redisCacheService.delete(topicKey);
      redisCacheService.delete(futureKey);

    }


}
